/*
 * Copyright 2020-2025 the original author or authors.
 * You cannot use this file unless authorized by the author.
 */
package org.ipig.messaging.kafka.producer;

import org.ipig.commons.conf.kafka.producer.KafkaProducerConf;
import org.ipig.commons.service.ReporterService;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;

import java.text.MessageFormat;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * AbstractKafkaProducer
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id: AbstractKafkaProducer.java 3 2020-01-21 09:36:33Z ts <comchnts@163.com> $$
 * @since 1.0
 */
@Slf4j
public abstract class AbstractKafkaProducer implements KafkaProducerService {
    /**
     * kafka生产者配置属性
     */
    @Setter
    @Getter
    private KafkaProducerConf conf;
    /**
     * 汇报器服务
     */
    @Setter
    @Getter
    private ReporterService<String, Void> reporterService;
    /**
     * 处理的消息总数量
     */
    @Getter
    private volatile LongAdder total = new LongAdder();
    /**
     * 生产者
     */
    private volatile KafkaProducer<String, String> producer;
    /**
     * 开始时间 时间戳
     */
    @Getter
    @Setter
    private volatile long beginTimeMills;


    /**
     * 获取生产者
     *
     * @return
     */
    public KafkaProducer<String, String> getProducer() {
        if (producer == null) {
            synchronized (this) {
                if (producer == null) {
                    Properties props = new Properties();
                    props.put("bootstrap.servers", conf.getBootstrapServers());
                    props.put("retries", conf.getRetries());
                    props.put("batch.size", conf.getBatchSize());
                    props.put("linger.ms", conf.getLingerMs());
                    props.put("buffer.memory", conf.getBufferMemory());
                    props.put("key.serializer", conf.getKeySerializer());
                    props.put("value.serializer", conf.getValueSerializer());
                    props.put("max.in.flight.requests.per.connection", "1");
                    if (StringUtils.equalsIgnoreCase("true", conf.getEnableIdempotence())) {
                        props.put("enable.idempotence", true);
                        props.put("acks", "all");
                        //当幂等性开启的时候acks即为all。如果显性的将acks设置为0，-1，那么将会报错Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence.
                    } else {
                        props.put("enable.idempotence", false);
                        props.put("acks", conf.getAcks());
                    }
                    ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
                    service.scheduleAtFixedRate(new Runnable() {
                        public void run() {
                            String msg = MessageFormat.format("start time Millis={}.reporting time Millis={},total quantity={}", beginTimeMills, System.currentTimeMillis(), total.longValue());
                            if (reporterService != null) {
                                reporterService.report(msg);
                            } else {
                                log.info(msg);
                            }
                        }
                    }, 10, 30, TimeUnit.SECONDS);
                    producer = new KafkaProducer<String, String>(props);
                }
            }
        }
        return producer;
    }
}
